package com.intct.flink.project.dim;

import com.intct.common.FlinkSqlWithUtil;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-16 09:05
 */
public class DimOrderStateSQLApp {
    public static void main(String[] args) throws Exception {
        // 创建Stream环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建Table环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 设置全局并行度
        env.setParallelism(1);

        // 开启检查
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 映射MySQL订单状态表
        tenv.executeSql(
                "CREATE TABLE order_state(                " +
                        "     id        INT                        " +
                        "     ,state    STRING                     " +
                        "     ,name     STRING                     " +
                        "     ,PRIMARY KEY (id) NOT ENFORCED       " +
                        " )                                        " +
                        FlinkSqlWithUtil.getMysqlCdcWith("travel", "order_state", "initial")
        );

        // 映射HBase订单状态维表
        tenv.executeSql(
                "CREATE TABLE dim_order_state(                     " +
                        "    id    STRING                                   " +
                        "    ,info  ROW<state       STRING                  " +
                        "			 ,name         STRING>                  " +
                        ") " +
                        FlinkSqlWithUtil.getHBaseSinkWith("dim", "dim_order_state")
        );

        // 将维表数据插入HBase
        tenv.executeSql(
                "INSERT INTO dim_order_state      " +
                        "SELECT                            " +
                        "    CAST(id AS STRING)            " +
                        "	,ROW(state                     " +
                        "	,name) AS info                 " +
                        "FROM                              " +
                        "    order_state                   "
        );
    }
}
